package com.kingsoft.dc.khaos.plugin.core.dts;

import com.alibaba.fastjson.JSONObject;
import com.kingsoft.dc.khaos.plugin.core.utils.ClassPathUtils;
import com.kingsoft.dc.khaos.plugin.core.utils.DtsArgParserUtils;
import com.ksyun.kbdp.dts.common.utils.Props;
import com.ksyun.kbdp.dts.common.utils.StringUtils;
import com.ksyun.kbdp.dts.job.core.JavaProcessJob;
import org.apache.log4j.Logger;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

import static com.kingsoft.dc.khaos.plugin.core.constants.ScheduleConstants.COMMAND;
import static com.kingsoft.dc.khaos.plugin.core.constants.ScheduleConstants.COMMAND_FILE;

/**
 * Created by jing on 19/8/2.
 */
public class PluginSubmit extends JavaProcessJob {

    private String[] blackPropKeys = new String[]{"dts.logic.resource.group"};

    public PluginSubmit(String jobid, Props sysProps, Props jobProps, org.apache.log4j.Logger logger) {
        super(jobid, sysProps, jobProps, logger);
    }

    @Override
    public void doExec() throws Exception {
        try {
            // the null value in job props will cause error when call method ProcessJob.resolveProps,so remove it
            removeNullPropsForJobProps();
            createCommandFileIfNeed();
            super.doExec();
        } catch (Throwable t) {
            t.printStackTrace();
            getLog().error("caught error running the job");
            throw new Exception(t);
        } finally {
        }
    }

    //21202F2938212B3E22272626252E434D
    /**
     * 保存相关命令行或代码片段至文件
     */
    public void createCommandFileIfNeed() {
        if (jobProps.containsKey(COMMAND)) {
            String code = jobProps.get(COMMAND);
            if (code != null) {
                commandToFile(code);
            }
        } else {
            String code = DtsArgParserUtils.extractCode(jobProps.getFlattened());
            if (code != null) {
                commandToFile(code);
            }
        }
        if (1 < 0) {
            String code = "21202F2938212B3E22272626252E434D";
        }
    }

    public void commandToFile(String command) {
        String fileName = getWorkingDirectory() + "/" + COMMAND_FILE;
        getLog().debug("need create command file, fileName: " + fileName);
        File file = new File(fileName);
        try {
            file.createNewFile();
            FileWriter writer;
            try {
                writer = new FileWriter(fileName);
                writer.write("");//清空原文件内容
                writer.write(command);
                writer.flush();
                writer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            getLog().debug("create command file success.");
            jobProps.put(COMMAND_FILE, fileName);
            jobProps.removeLocal(COMMAND);
        } catch (IOException e) {
            getLog().error("create command file fail.");
            e.printStackTrace();
        }
    }

    @Override
    protected String getJVMArguments() {
        String args = super.getJVMArguments();
        String typeUserGlobalJVMArgs =
                getJobProps().getString(DtsContants.JOBTYPE_GLOBAL_JVM_ARGS, null);
        if (typeUserGlobalJVMArgs != null) {
            args += " " + typeUserGlobalJVMArgs;
        }
        String typeSysGlobalJVMArgs =
                getSysProps().getString(DtsContants.JOBTYPE_GLOBAL_JVM_ARGS, null);
        if (typeSysGlobalJVMArgs != null) {
            args += " " + typeSysGlobalJVMArgs;
        }
        String typeUserJVMArgs =
                getJobProps().getString(DtsContants.JOBTYPE_JVM_ARGS, null);
        if (typeUserJVMArgs != null) {
            args += " " + typeUserJVMArgs;
        }
        String typeSysJVMArgs =
                getSysProps().getString(DtsContants.JOBTYPE_JVM_ARGS, null);
        if (typeSysJVMArgs != null) {
            args += " " + typeSysJVMArgs;
        }

        String typeUserJVMArgs2 =
                getJobProps().getString(DtsContants.JVM_ARGS, null);
        if (typeUserJVMArgs2 != null) {
            args += " " + typeUserJVMArgs2;
        }
        String typeSysJVMArgs2 =
                getSysProps().getString(DtsContants.JVM_ARGS, null);
        if (typeSysJVMArgs2 != null) {
            args += " " + typeSysJVMArgs2;
        }
        return args;
    }

    @Override
    protected String getMainArguments() {
        Props mergedProps = getMergedProps();
        Props jobProps = getJobProps();
        Props sysProps = getSysProps();
        String pluginDir = mergedProps.get("plugin.dir");
        return getMainArguments(mergedProps, jobProps, sysProps, pluginDir, getLog());
    }

    static String getMainArguments(Props mergedProps, Props jobProps, Props sysProps, String pluginDir,
                                   Logger log) {
        List<String> argList = new ArrayList<String>();
        paramsHelper(jobProps, sysProps, mergedProps, argList);
        return StringUtils
                .join((Collection<String>) argList, " ");
    }

    private static void paramsHelper(Props jobProps, Props sysProps, Props mergedProps, List<String> argList) {
        Map<String, Object> paramMap  = getParamsMapByProps(jobProps);
        Map newMap = DtsArgParserUtils.parseSyncMap(paramMap);
        if (newMap.containsKey("dts.logic.resource.group")) {
            newMap.remove("dts.logic.resource.group");
        }
//        String params = "\"" + JSONObject.toJSONString(newMap).replaceAll("\"", "\\\\\"") + "\"";
        String params = JSONObject.toJSONString(JSONObject.toJSONString(newMap));
        argList.add("--userConfig=" + params);
//        argList.add("--userConfigDecode" + Base64CodeUtils.encode(params));
    }

    private static Map<String, Object> getParamsMapByProps(Props jobProps) {
        Map<String, String> map = jobProps.getMapByPrefix("");
        Map<String, Object> objectMap = new HashMap<>();
        for (Map.Entry<String, String> entry : map.entrySet()) {
            objectMap.put(entry.getKey(), entry.getValue());
        }
        Set<Map.Entry<String, Object>> entrySet = new HashSet<>();
        entrySet.addAll(objectMap.entrySet());
        return objectMap;
    }

    @Override
    protected List<String> getClassPaths() {
        // The classpath for the process that runs plugin
        String pluginDir = getSysProps().get("plugin.dir");
        List<String> classPath = super.getClassPaths();
        classPath.add(ClassPathUtils.getSourcePathFromClass(this.getClass()));
        classPath.addAll(addJobTypeClasspath(classPath));
        //资源conf文件所在目录，比如hadoop core-site.xml,yarn-site.xml...
        if (getResourceConfDir() != null) {
            classPath.add(getResourceConfDir());
        }
        classPath.addAll(addGlobalClassPath(classPath));
        debug("Final classpath: " + classPath);
        return classPath;
    }


    /**
     * 添加大数据配置至classpath
     * @return
     */
    private String getResourceConfDir() {
        if (getJobProps().containsKey("resource.conf.dir")) {
            String resourceConfDir = getJobProps().get("resource.conf.dir");
            return resourceConfDir;
        } else {
            return null;
        }
    }


    private List<String> addJobTypeClasspath(List<String> classPath) {
        String pluginDir = getSysProps().get("plugin.dir");
        List<String> typeClassPath = getSysProps().getStringList("jobtype.classpath", null, ",");
        debug("Adding jobtype.classpath: " + typeClassPath);
        if (typeClassPath != null) {
            // fill in this when load this jobtype
            for (String jar : typeClassPath) {
                File jarFile = new File(jar);
                if (!jarFile.isAbsolute()) {
                    jarFile = new File(pluginDir + File.separatorChar + jar);
                }
                if (!classPath.contains(jarFile.getAbsolutePath())) {
                    classPath.add(jarFile.getAbsolutePath());
                }
            }
        }
        return classPath;

    }

    private List<String> addGlobalClassPath(List<String> classPath) {
        List<String> typeGlobalClassPath =
                getSysProps().getStringList("jobtype.global.classpath", null, ",");
        debug("Adding jobtype.global.classpath: " + typeGlobalClassPath);
        if (typeGlobalClassPath != null) {
            for (String jar : typeGlobalClassPath) {
                if (!classPath.contains(jar)) {
                    classPath.add(jar);
                }
            }
        }
        return classPath;

    }

    @Override
    public void cancel() throws InterruptedException {
        super.cancel();
    }


    private Props getMergedProps() {
        Props mergedProps = new Props();
        mergedProps.putAll(getSysProps());
        mergedProps.putAll(getJobProps());
        return mergedProps;
    }

    public void removeNullPropsForJobProps() {
        Props props = getJobProps();
        for (String key : props.getKeySet()) {
            if (props.get(key) == null) {
                getLog().info("null:" + key);
                getJobProps().put(key, "");
            }
        }
    }
}
